package com.mmc.delay.timer;


import com.mmc.delay.api.DelayFacade;
import com.mmc.delay.config.DelayCoreConfig;
import com.mmc.delay.util.LockUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class DelayMoveToReadyTimer {

    private DelayFacade delayFacade;

    private volatile boolean moveStop=false;

    public DelayMoveToReadyTimer(DelayFacade delayFacade) {
        this.delayFacade=delayFacade;
    }

    private DelayMoveToReadyTimer(){}

    private static final Logger logger = LoggerFactory.getLogger(DelayMoveToReadyTimer.class);

    private static final Long MOVE_REDIS_LOCK_TIME=3L;

    public static final ThreadPoolExecutor DELAY_REMOVE_THREAD_POOL_EXECUTOR=new ThreadPoolExecutor(DelayCoreConfig.getPartitionCount(),DelayCoreConfig.getPartitionCount(),0, TimeUnit.SECONDS,new LinkedBlockingDeque<Runnable>(1000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread=new Thread(r);
            thread.setName("delay-remove-thread");
            return thread;
        }
    });


    public static String getMoveLockKey(String topic){
        return "delay:"+topic;
    }



    public synchronized void runMove(){
       Integer partitionCount=DelayCoreConfig.getPartitionCount();
        for (int i = 1; i <= partitionCount; i++) {
            final int finalI = i;
            DELAY_REMOVE_THREAD_POOL_EXECUTOR.submit(new Runnable() {
                @Override
                public void run() {
                    while (!moveStop){
                        doRun(String.valueOf(finalI));
                    }
                }
            });
        }
    }

    private void doRun(String partition){
        long time = LockUtil.getNextTime(partition).get() - System.currentTimeMillis();
        if (time > 0) {
            try {
                synchronized (LockUtil.LOCK) {
                    LockUtil.LOCK.wait(time);
                }
            } catch (InterruptedException e) {
                logger.error("移动delay Bucket线程中断",e);
            }
        }else {
            String lockKey=getMoveLockKey(partition);
            boolean lock=false;
            String uuid= UUID.randomUUID().toString();
            try {
                lock = delayFacade.lockTimer(lockKey,uuid,MOVE_REDIS_LOCK_TIME);
                if(lock){
                    long runTime = delayFacade.moveAndRtNextRunTime(partition);
                    LockUtil.getNextTime(partition).set(runTime);
                }
            }finally {
                if(lock){
                    delayFacade.unlock(lockKey,uuid);
                }
            }


        }
    }
}
